/*
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.facebook.presto.obs;

import com.facebook.presto.hive.HiveClientConfig;
import com.obs.services.BasicObsCredentialsProvider;
import com.obs.services.IObsCredentialsProvider;
import com.obs.services.ObsClient;
import com.obs.services.ObsConfiguration;
import io.airlift.units.Duration;
import org.apache.hadoop.conf.Configuration;

import javax.annotation.concurrent.GuardedBy;
import java.util.Optional;

import static com.facebook.presto.obs.ObsConfigurationUpdater.OBS_ACCESS_KEY;
import static com.facebook.presto.obs.ObsConfigurationUpdater.OBS_CONNECT_TIMEOUT;
import static com.facebook.presto.obs.ObsConfigurationUpdater.OBS_ENDPOINT;
import static com.facebook.presto.obs.ObsConfigurationUpdater.OBS_MAX_CONNECTIONS;
import static com.facebook.presto.obs.ObsConfigurationUpdater.OBS_MAX_ERROR_RETRIES;
import static com.facebook.presto.obs.ObsConfigurationUpdater.OBS_PATH_STYLE_ACCESS;
import static com.facebook.presto.obs.ObsConfigurationUpdater.OBS_SECRET_KEY;
import static com.facebook.presto.obs.ObsConfigurationUpdater.OBS_SECURITY_TOKEN;
import static com.facebook.presto.obs.ObsConfigurationUpdater.OBS_SOCKET_TIMEOUT;
import static com.facebook.presto.obs.ObsConfigurationUpdater.OBS_SSL_ENABLED;
import static com.google.common.base.Strings.isNullOrEmpty;
import static com.google.common.base.Verify.verify;

public class PrestoObsClientFactory
{
    @GuardedBy("this")
    private ObsClient obsClient;

    synchronized ObsClient getObsClient(Configuration config,HiveClientConfig clientConfig)
    {
        if (obsClient != null) {
            return obsClient;
        }

        HiveObsConfig defaults = new HiveObsConfig();
        int maxErrorRetries = config.getInt(OBS_MAX_ERROR_RETRIES, defaults.getObsMaxErrorRetries());
        boolean sslEnabled = config.getBoolean(OBS_SSL_ENABLED, defaults.isObsSslEnabled());
        boolean isPathStyleAccess = config.getBoolean(OBS_PATH_STYLE_ACCESS, defaults.isObsPathStyleAccess());
        Duration connectTimeout = Duration.valueOf(config.get(OBS_CONNECT_TIMEOUT, defaults.getObsConnectTimeout().toString()));
        Duration socketTimeout = Duration.valueOf(config.get(OBS_SOCKET_TIMEOUT, defaults.getObsSocketTimeout().toString()));
        int maxConnections = config.getInt(OBS_MAX_CONNECTIONS, defaults.getObsMaxConnections());

        String endpoint = config.get(OBS_ENDPOINT);
        verify(endpoint != null,
                "Invalid configuration: endpoint must be set");

        ObsConfiguration obsConfiguration = new ObsConfiguration();
        obsConfiguration.setEndPoint(endpoint);
        obsConfiguration.setConnectionTimeout((int) connectTimeout.toMillis());
        obsConfiguration.setSocketTimeout((int) socketTimeout.toMillis());
        obsConfiguration.setMaxConnections(maxConnections);
        obsConfiguration.setMaxErrorRetry(maxErrorRetries);
        obsConfiguration.setPathStyle(isPathStyleAccess);

        IObsCredentialsProvider credentialsProvider = getCredentialsProvider(config,defaults);

        obsClient = new ObsClient(credentialsProvider,obsConfiguration);
        return obsClient;
    }


    private static IObsCredentialsProvider getCredentialsProvider(Configuration conf, HiveObsConfig defaults)
    {
        Optional<IObsCredentialsProvider> credentials = getObsCredentials(conf);
        if (credentials.isPresent()) {
            return credentials.get();
        }
        return null;
    }

    private static Optional<IObsCredentialsProvider> getObsCredentials(Configuration conf)
    {
        String accessKey = conf.get(OBS_ACCESS_KEY);
        String secretKey = conf.get(OBS_SECRET_KEY);

        if (isNullOrEmpty(accessKey) || isNullOrEmpty(secretKey)) {
            return Optional.empty();
        }
        String securityToken = conf.get(OBS_SECURITY_TOKEN);
        if (!isNullOrEmpty(securityToken)) {
            return Optional.of(new BasicObsCredentialsProvider(accessKey, secretKey, securityToken));
        }
        return Optional.of(new BasicObsCredentialsProvider(accessKey, secretKey));
    }

}
